package wsz;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;

/**
 * @author wsz
 * @date 2022/3/16 15:57
 **/
public class MyProducer {
    public static void main(String[] args) throws Exception {
        Map<String, Object> configs = new HashMap<>();
        configs.put("bootstrap.servers", "192.168.174.138:9092");
        configs.put("key.serializer", IntegerSerializer.class);
        configs.put("value.serializer", StringSerializer.class);

        // TODO 指定分区器
        configs.put("partitioner.class", MyPartitioner.class);


        KafkaProducer<Integer, String> producer = new KafkaProducer<>(configs);
        List<Header> headerList = new ArrayList<>();
        headerList.add(new RecordHeader("biz.name", "aaaa".getBytes()));

        ProducerRecord<Integer, String> record = new ProducerRecord<>(
                "topic_1",
                0,
                0,
                "message 0",
                headerList
        );

        // 同步发送
        Future<RecordMetadata> send = producer.send(record);
        RecordMetadata metadata = send.get();
        System.out.println(metadata.topic());
        System.out.println(metadata.partition());
        System.out.println(metadata.offset());
        producer.close();
    }
}
